package com.chenjj.bigdata.flink.book

import com.chenjj.bigdata.flink.window.function.{MyAverageAggregation, MyProcessWindowFunction}
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.{SlidingEventTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.junit.{Assert, Test}


/**
  * 第4.3章：Window
  */
@Test
class FlinkTesterChapter4_3 extends  Assert{
  val env = StreamExecutionEnvironment.getExecutionEnvironment

  /**
    * 测试 AggregateFunction: MyAverageAggregation
    * 求每个窗口的平均值
    */
  @Test
  def test1(): Unit ={
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val data = List(("a",1L),("b",1L),("c",3L),("a",4L),("a",5L),("a",6L),("d",5000L),("a",3000L))
    val inputStream = env.fromCollection(data)

    //滚动窗口
    inputStream
      .assignAscendingTimestamps(t=>t._2)
      .keyBy(_._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(1)))
      .aggregate(new MyAverageAggregation())
      .map(x=>{println("平均值:"+x); x})

    env.execute()

    //输出
    //平均值:1.0   - 分区b [0,1000)的平均值
    //平均值:4.0   - 分区a  [0,1000)的平均值
    //平均值:3.0   - 分区c  [0,1000)的平均值
    //平均值:5000.0  -分区d [5000-6000)的平均值
    //平均值:3000.0  -分区a [3000,4000)的平均值
  }

  /**
    * 测试ProcessWindowFunction
    * 求每个窗口的平均值
    */
  @Test
  def test2():Unit={
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val data = List(("a",1L,10),("b",1L,15),("c",3L,3),("a",4L,29),("a",5L,100),("a",6L,4),("d",5000L,5),("a",3000L,5))
    val inputStream = env.fromCollection(data)

    inputStream
      .assignAscendingTimestamps(t=>t._2)
      .keyBy(_._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(1)))
      .process(new MyProcessWindowFunction)
      .print()

    env.execute()
    //输出
    // key,最小值，最大值，求和，平均值，窗口start，窗口end
    //3> (b,15,15,15,15,0,1000)
    //7> (d,5,5,5,5,5000,6000)
    //6> (c,3,3,3,3,0,1000)
    //8> (a,4,100,143,35,0,1000)
    //8> (a,5,5,5,5,3000,4000)
  }


  /**
    * 测试ReduceFunction+ProcessWindowFunction: 统计每个分区的最小值并打印窗口信息
    *
    * ReduceFunction计算最小值
    *ProcessWindowFunction拿到每个窗口的计算结果
    */
  @Test
  def test3:Unit={
    val data = List(("a",1L,10),("b",1L,15),("c",3L,3),("a",4L,29),("a",5L,100),("a",6L,4),("d",5000L,5),("d",7000L,3))
    val inputStream = env.fromCollection(data)

    inputStream
      .assignAscendingTimestamps(t=>t._2)
      .keyBy(_._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(1)))
      .reduce(
        (r1:(String,Long,Int),r2:(String,Long,Int))=>{
          if(r1._3>r2._3) r2 else r1
        },
        (key:String,window:TimeWindow,minReadings:Iterable[(String,Long,Int)],out:Collector[(String,Long,Long,(String,Long,Int))])=>{
            val it = minReadings.iterator
            while(it.hasNext){
              println("["+window.getStart+","+window.getEnd+"):"+it.next())
            }
            out.collect(key,window.getStart,window.getEnd,minReadings.iterator.next())
        }
      )

    env.execute()
    //输出
    //[5000,6000):(d,5000,5)
    //[0,1000):(b,1,15)
    //[0,1000):(c,3,3)
    //[0,1000):(a,6,4)
    //
    //7> (d,5000,6000,(d,5000,5))
    //8> (a,0,1000,(a,6,4))
    //6> (c,0,1000,(c,3,3))
    //3> (b,0,1000,(b,1,15))
  }


}
